/*
 * main.cpp
 *
 *  Created on: 2011-7-3
 *      Author: yong
 */
#include<boost/mpi.hpp>
#include <boost/program_options.hpp>
#include <boost/program_options/parsers.hpp>
#include <cpuinfo/cpuinfo.h>
#include "MemPageMaster.h"
#include "GMMMapper.h"
#include "GMMInitMapper.h"
#include "Reducer.h"
#include "GMMReducer.h"
#include "math/MatrixUtil.h"
#include "Typedef.h"
using namespace boost::algorithm;
using namespace distrim;
namespace po = boost::program_options;
using boost::mpi::communicator;
using boost::mpi::environment;
typedef boost::program_options::options_description OptionsDescription;
typedef boost::program_options::variables_map VariablesMap;
typedef boost::program_options::command_line_parser CommandLineParser;
typedef boost::program_options::error Error;
int main(int argc, char **argv)
{
	environment env(argc, argv);
	communicator world;
	if (world.size() < 2)
	{
		std::cout << "Process requirement: >= 2." << std::endl;
		env.abort(-1);
	}
	Config config;
	int cpuCount = cpuinfo_processor_count(-1);
	if (world.rank() == 0) // Starting reducer process.
	{
		try
		{
			OptionsDescription desc("Allowed option");
			VariablesMap vm;
			desc.add_options()("help,h", "print usage message")("input,i",
					po::value<std::string>()->required(), "path name for input")(
					"output,o", po::value<std::string>()->required(),
					"path name for output file")("clusterNumber,c", po::value<
					size_t>()->required(), "number of clusters")(
					"convergenceDelta,e", po::value<long double>()->required(),
					"convergence delta for each point")("maxIteration,x",
					po::value<size_t>()->required(), "maximum iteration number")(
					"dimension,d", po::value<size_t>()->required(),
					"dimension of points")("ioThreads,t",
					po::value<size_t>()->default_value(cpuCount),
					"io thread in bounded-cache mode, no effect in full-cache mode")(
					"preTraining,p",
					po::value<std::string>()->default_value("")->required(),
					"Pre-training input in semi-supervised clustering");
			if (cpuCount != -1)
			{
				desc.add_options()("mapperNumber,m",
						po::value<size_t>()->default_value(cpuCount),
						"mapper threads within a process")("bufferNumber,b",
						po::value<size_t>()->default_value(cpuCount),
						"input buffer for mappers within a process")(
						"fileSplits,s", po::value<size_t>()->default_value(
								cpuCount),
						"splits for input file, equal to buffer number if in full-cache mode.");
			}
			else
			{
				desc.add_options()("mapperNumber,m",
						po::value<size_t>()->required(),
						"mapper threads within a process")("bufferNumber,b",
						po::value<size_t>()->required(),
						"input buffer for mappers within a process")(
						"fileSplits,s", po::value<size_t>()->required(),
						"splits for input file, equal to buffer number if in full-cache mode.");
			}
			store(parse_command_line(argc, argv, desc), vm); // Could throw exceptions.
			world.barrier(); // #1;
			for (int i = 1; i < world.size(); ++i)
			{
				world.send(i, msg_cmd_tag, (int) option_good);
			}
			/////////////////////////////////////
			if (vm.count("help"))
			{
				world.barrier();
				for (int i = 1; i < world.size(); ++i)
				{
					world.send(i, msg_cmd_tag, (int) option_need_help);
				}
				std::cout << desc << std::endl;
				return 0;
			}
			else
			{
				notify(vm); // Could throw  exceptions.
				world.barrier(); // #2
				for (int i = 1; i < world.size(); ++i)
				{
					world.send(i, msg_cmd_tag, (int) option_good);
				}
			}
			config.SetClusterNum(vm["clusterNumber"].as<size_t> ());
			config.SetConvergeDelta(vm["convergenceDelta"].as<long double> ());
			config.SetFileSplitsCount(vm["fileSplits"].as<size_t> ());
			config.SetIOThreadCount(vm["ioThreads"].as<size_t> ());
			config.SetInputBufferCount(vm["bufferNumber"].as<size_t> ());
			config.SetInputPath(vm["input"].as<std::string> ());
			config.SetOutputFileName(vm["output"].as<std::string> ());
			config.SetMapperCount(vm["mapperNumber"].as<size_t> ());
			config.SetMaxIterCount(vm["maxIteration"].as<size_t> ());
			config.SetValueElemCount(vm["dimension"].as<size_t> ());
			config.SetIOThreadCount(vm["ioThreads"].as<size_t> ()); // If cast failed, send msg to notify slaves to exit in catch.
			config.SetPreTrainingFile(vm["preTraining"].as<std::string>());
			world.barrier(); // #3
			for (int i = 1; i < world.size(); ++i)
			{
				world.send(i, msg_cmd_tag, (int) option_good);
			}
			world.barrier(); // #4, get ready to send parsed configurations.
			for (int i = 1; i < world.size(); ++i)
			{
				world.send(i, msg_config_tag, config);
			}
			size_t totalPoints = 0;
			world.barrier();
			mpi::all_reduce(world, totalPoints, totalPoints,
					std::plus<size_t>());
			config.SetTotalPoints(totalPoints);
			GMMReducer<int, VALUE_TYPE> gr(config);
			gr();
		} catch (const Error &e) // If missing options
		{
			// catch #1 or #2.
			world.barrier(); // Any exceptions occured due to incomplete options, notify all slaves to exit.
			for (int i = 1; i < world.size(); ++i)
			{
				world.send(i, msg_cmd_tag, (int) option_logic_error); // Cast to int is important,
				// or it'll send sizeof(option_logic_error), that'll always be 4.
			}
			std::cout << e.what() << std::endl;
		} catch (const boost::bad_any_cast &e) // If option value is invalid.
		{
			world.barrier(); // catch #3
			for (int i = 1; i < world.size(); ++i)
			{
				world.send(i, msg_cmd_tag, (int) option_bad_cast_error);
			}
			std::cout << e.what() << std::endl;
		} catch (const std::exception &e)
		{
			std::cout << e.what() << std::endl;
			env.abort(-2);
		} catch (...)
		{
			env.abort(-3);
		}
	}
	else
	{
		int cmdOptionMsg = 0;
		world.barrier(); // #1
		world.recv(0, msg_cmd_tag, cmdOptionMsg);
		if (cmdOptionMsg == option_logic_error)
		{
			return 0;
		}
		/////////////options stored ok./////////////////////
		world.barrier(); // #2, waiting for the command line to be parsed.
		world.recv(0, msg_cmd_tag, cmdOptionMsg); // See if options are complet or should print help
		// infomation.
		if (cmdOptionMsg == option_need_help)
		{
			return 0;
		}
		else if (cmdOptionMsg == option_logic_error)
		{
			return 0;
		}
		////////////options are complete./////////////////
		world.barrier(); // #3, wating for the values to be parsed.
		world.recv(0, msg_cmd_tag, cmdOptionMsg);
		if (cmdOptionMsg == option_bad_cast_error) // If option values cast failed in master, exit.
		{
			return 0;
		}
		world.barrier(); // #4, waiting to receive configurations.
		world.recv(0, msg_config_tag, config);
		size_t totalPoints = 0;
		fs::path inputPath(config.GetInputPath());
		size_t procTotalPoints = Util::CountLines(inputPath);
		config.SetLocalPoints(procTotalPoints);
		size_t splits = config.GetFileSplitsCount();
		world.barrier(); // Wating for the total number of points being summated.
		mpi::all_reduce(world, procTotalPoints, totalPoints,
				std::plus<size_t>());
		if (procTotalPoints % splits == 0)
		{
			config.SetPageCapacity(procTotalPoints / splits);
		}
		else
		{
			config.SetPageCapacity((procTotalPoints / splits + 1));
		}
		config.SetTotalPoints(totalPoints);
		MemPageMaster<void, double, int, VALUE_TYPE> memPageMaster(config); // Start mappers.
	}

	return 0;
}
